package cn.tedu.batch.transformation

import org.apache.flink.api.scala.ExecutionEnvironment

/**
 * @author Amos
 * @date 2022/5/19
 */
// flatMap 算子，过程为先map,然后flat处理
object BatchFlatMapDemo {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment

    import org.apache.flink.api.scala._
    val source = env.fromElements("张三,中国,江西省,南昌市",
      "李四,中国,河北省,石家庄市",
      "Tom,America,NewYork,Manhattan")

    val rst = source.flatMap(elem => {
      val infos = elem.split(",")
      val res1 = infos(0) + "," + infos(1)
      val res2 = infos(0) + "," + infos(1) + infos(2)
      val res3 = infos(0) + "," + infos(1) + infos(2) + infos(3)
      val lst = List(res1, res2, res3)
      lst
    })

    rst.print()
  }
}
